package com.iwhalecloud.uncc.mq.producer;
import java.nio.charset.StandardCharsets;
import java.util.List;
import java.util.UUID;

import javax.annotation.PreDestroy;

import org.apache.commons.lang.exception.ExceptionUtils;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.MessageQueueSelector;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageQueue;

import com.iwhalecloud.uncc.model.MqClusterConfig;
import com.iwhalecloud.uncc.utils.ZipUtil;

import lombok.extern.slf4j.Slf4j;

@Slf4j
public class RocketMqProducer implements MqProducer {

    private DefaultMQProducer producer;

    @Override
    public void init(MqClusterConfig mqConfig) {
        try {
            if (log.isInfoEnabled()) {
                log.info("producer init...");
            }
            String producerGroup = UUID.randomUUID().toString();
            producer = new DefaultMQProducer(producerGroup);
            producer.setNamesrvAddr(mqConfig.getNameserv());
            producer.setInstanceName(producerGroup);
            producer.setSendMsgTimeout(mqConfig.getSendTimeout());
            producer.start();

            if (log.isInfoEnabled()) {
                log.info("producer init success!");
            }
        } catch (Exception e) {
            log.error("producer init exception!"+ ExceptionUtils.getStackTrace(e));
        }
    }

    @Override
    @PreDestroy
    public void destroy() {
        if (producer != null) {
            producer.shutdown();
        }
    }

    /**
     * 发送普通消息
     *
     * @throws Exception
     */
    @Override
    public Object sendMsg(String topic, String tag, String key, String msg) throws Exception {
       return this.sendMsg(topic, tag, key, msg, true);
    }

    /**
     * 发送顺序消息
     */
    @Override
    public Object sendOrderlyMsg(String topic, String tag, String key, String msg, String orderlyKey) throws Exception {
        return this.sendOrderlyMsg(topic, tag, key, msg, orderlyKey, true);
    }

    /**
     * 发送普通消息
     *
     * @throws Exception
     */
    @Override
    public Object sendMsg(String topic, String tag, String key, String msg, boolean compress) throws Exception {
       Message message = new Message(topic, tag, key,compress ? ZipUtil.zip(msg.getBytes(StandardCharsets.UTF_8)) : msg.getBytes(StandardCharsets.UTF_8));
        SendResult sendResult = producer.send(message);
        return sendResult;
    }

    @Override
    public Object sendMsg(String topic, String tag, String key, byte[] msgArr ) throws Exception {
        Message message = new Message(topic, tag, key,  msgArr);
        SendResult sendResult = producer.send(message);
        return sendResult;
    }


    /**
     * 发送顺序消息
     */
    @Override
    public Object sendOrderlyMsg(String topic, String tag, String key, String msg, String orderlyKey,
                                 boolean compress) throws Exception {
    	int orderly = getOrderlyHashCode(orderlyKey);
    	
        Message message = new Message(topic, tag, key, compress ? ZipUtil.zip(msg.getBytes(StandardCharsets.UTF_8)) : msg.getBytes(StandardCharsets.UTF_8));
        SendResult sendResult = producer.send(message, new MessageQueueSelector() {
            @Override
            public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
				//modified by hanpengfei 20211015,要根据orderlykey来选队列而不是根据消息
//                int index =  Math.abs(msg.hashCode()) % mqs.size();
//                return mqs.get(index);
            	int id = Math.abs((int) arg);
				int index = id % mqs.size();
				return mqs.get(index);
            }
        }, orderly);
        return sendResult;
    }
    
	private int getOrderlyHashCode(String id) {
		int key = id.hashCode();
		// 可能产生最小值，所以处理一下
		if (key != Integer.MIN_VALUE) {
			return Math.abs(key);
		} else {
			return 0;
		}
	}
}
